package com.wj.opratorlog.cdc.debezium;/**
 * @author
 * @description
 * @create 2024-08-24-18:47
 **/




import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import java.io.File;
import java.io.IOException;
import java.nio.file.Paths;


@Configuration
public class DebeziumConnectorConfig {


    @Bean

    public io.debezium.config.Configuration customerConnector( Environment env) throws IOException {

        File offsetStorageTempFile =   File.createTempFile("offsets_", ".dat", new File("D:\\cdcdata"));
        File hsiotryStoreTemplete = File.createTempFile("hisotry_", ".dat", new File("D:\\cdcdata"));


        return io.debezium.config.Configuration.create()
                .with("name", "customer_mysql_connector")
                .with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
                .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
                .with("offset.storage.file.filename", offsetStorageTempFile.getAbsolutePath())
                .with("connectionTimeZone",  "UTC")
                .with("offset.flush.interval.ms", "60000")
                .with("database.hostname", env.getProperty("customer.datasource.host"))
                .with("database.port", env.getProperty("customer.datasource.port")) // defaults to 5432
                .with("database.user", env.getProperty("customer.datasource.username"))
                .with("database.password", env.getProperty("customer.datasource.password"))
                .with("database.dbname", env.getProperty("customer.datasource.database"))
              //  .with("table.include.list", "your_database.your_table_name")

                .with("database.server.id", env.getProperty("customer.datasource.serverid"))
                .with("database.server.name", "msqlcdc")

                //历史变更记录
                 .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
                 .with("database.history.file.filename", hsiotryStoreTemplete.getAbsolutePath())

                    //格式化日期 日期处理问题（时区处理问题）
                .with("converters", "dateConverters")
                .with("dateConverters.type", "com.wj.opratorlog.cdc.debezium.MySqlDateTimeConverter")
                .with("dateConverters.format.date", "yyyy-MM-dd")
                .with("dateConverters.format.time", "HH:mm:ss")
                .with("dateConverters.format.datetime", "yyyy-MM-dd HH:mm:ss")
                .with("dateConverters.format.timestamp", "yyyy-MM-dd HH:mm:ss")
                .with("dateConverters.format.date", "yyyy-MM-dd")
                .with("dateConverters.format.timestamp.zone", "UTC+8")
            //全局读写锁,会影响在线业务,所以跳过锁设置
                .with("debezium.snapshot.locking.mode","none")
            //是否包含数据库表结构层面的变更,建议使用默认值true
              .with("include.schema.changes", "true")

//                props.setProperty("converters", "dateConverters");
//        props.setProperty("dateConverters.type", "com.zy.data.sync.common.conver.MySqlDateTimeConverter");
//        props.setProperty("dateConverters.format.date", "yyyy-MM-dd");
//        props.setProperty("dateConverters.format.time", "HH:mm:ss");
//        props.setProperty("dateConverters.format.datetime", "yyyy-MM-dd HH:mm:ss");
//        props.setProperty("dateConverters.format.timestamp", "yyyy-MM-dd HH:mm:ss");
//        props.setProperty("dateConverters.format.timestamp.zone", "UTC+8");


                . with("key.converter.schemas.enable", "false").
                 with("value.converter.schemas.enable", "true").
               // 是否包含数据库表结构层面的变更，建议使用默认值true
                 with("include.schema.changes", "true").
                with("tombstones.on.delete", "false")
               // .with("snapshot.mode", "always")  initial_only
                .build();
    }
}
